package com.atguigu.flink.chapter10;

import com.atguigu.flink.bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2023/6/27 09:23
 */
public class TableSQLDemo3 {
    public static void main(String[] args) throws Exception {
        // 1. 表的执行
        /*EnvironmentSettings conf = EnvironmentSettings.newInstance()
            .inStreamingMode()
            .build();
        TableEnvironment tEnv = TableEnvironment.create(conf);*/
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<WaterSensor> stream = env.fromElements(
            new WaterSensor("s1", 1000L, 10),
            new WaterSensor("s2", 2000L, 20),
            new WaterSensor("s1", 3000L, 30),
            new WaterSensor("s1", 4000L, 40),
            new WaterSensor("s2", 5000L, 50)
        );
    
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        Table table = tEnv.fromDataStream(stream);
        
        // 1. 查询未注册的表
//        tEnv.executeSql("ddl 增删改");
//        tEnv.sqlQuery("查询语句");
    
//        Table result = tEnv.sqlQuery("select * from " + table);
//        Table result = tEnv.sqlQuery("select id, sum(vc) vc_sum from " + table + " group by id");
        
    
        // 2. 查询已注册的表
        tEnv.createTemporaryView("sensor", table);
        tEnv.sqlQuery("select * from sensor").execute().print();
        
     //   result.execute().print();
    }
}
